package com.yan.yanbi.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

/**
 * 消息消费者[按轮询规则一人一条]
 *
 * @author 86177
 */
public class MultiConsumer {

    private static final String TASK_QUEUE_NAME = "multi_queue";

    public static void main(String[] argv) throws Exception {
        // 建立链接
        ConnectionFactory factory = new ConnectionFactory();
        // 设置 rabbitmq 对应的信息
        factory.setHost("47.109.81.239");
        factory.setUsername("yhnyhn123");
        factory.setPassword("1658450511");
        //从工厂中获取连接
        final Connection connection = factory.newConnection();

        for (int i = 0; i < 2; i++) {
            // 创建两个消费者频道
            final Channel channel = connection.createChannel();
            //声明队列
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            // 控制当个消费者的积压数，每个消费者处理一个消息
            channel.basicQos(1);

            int finalI = i;
            //创建消息接收回调函数，以便接收消息
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                // 获取消息内容将接受到的消息转化为字符串
                String message = new String(delivery.getBody(), "UTF-8");
                try {
                    // 处理工作
                    System.out.println(" [x] Received '消费者：" + finalI + "，消费了：" + message + "'");
                    // 指定取某条消息，确认消息
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    // 停20秒 模拟机器处理能力有限
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.getStackTrace();
                    // 第三个参数
                    //表示消费者拒绝接收消息，要求重新投递，消息失败
                    /*
                     * 第一个参数是消息的交付标签（delivery tag），
                     * 第二个参数表示是否将消息重新放入队列中（false 表示不重新放入），
                     * 第三个参数表示是否拒绝多条消息。通常用于处理无法处理的消息或者需要重新处理的消息。
                     */
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
                } finally {
                    System.out.println(" [x] Done");
                    //手动发送应答，告诉RabbitMQ消息已经被处理
                    /*
                     *这行代码会发送一个 acknowledgment (Ack) 给 RabbitMQ，表示消费者已成功处理并确认接收了消息。
                     * 第一个参数是消息的交付标签（delivery tag），
                     * 第二个参数表示是否一次性确认所有小于当前交付标签的消息。通常用于告知 RabbitMQ 消息已被正常处理，可以从队列中删除。
                     */
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };
            // 开启消费监听，会一直监听生产者的消息
            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });
        }
    }
}
